package com.deep.flink.mysql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class TableSync {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //创建用于输出的表
        String createInputDDL = "create table ny_energy_data_source\n" +
                "(\n" +
                "    id             bigint ,\n" +
                "    name  string     ,\n" +
                "    create_datetime       timestamp(0)   ,\n" +
                "    PRIMARY KEY (`id`) NOT ENFORCED\n" +
                ")WITH (\n" +
                "    'connector' = 'mysql-cdc',\n" +
                "    'hostname' = 'localhost',\n" +
                "    'port' = '3320',\n" +
                "    'username' = 'root',\n" +
                "    'password' = '123456',\n" +
                "    'database-name' = 'cdc_test',\n" +
                "    'table-name' = 'test1'\n" +
                ") ";
        tableEnv.executeSql(createInputDDL);


        //创建用于输出的表
        String createOutputDDL = "create table ny_energy_data_target\n" +
                "(\n" +
                "    id             bigint ,\n" +
                "    name  string     ,\n" +
                "    create_datetime       timestamp(0)   ,\n" +
                "    PRIMARY KEY (`id`) NOT ENFORCED\n" +
                ")WITH (\n" +
                "'connector' = 'jdbc',\n" +
                "'driver' = 'com.mysql.cj.jdbc.Driver',\n" +
                "'url' = 'jdbc:mysql://localhost:3320/cdc_test?rewriteBatchedStatements=true',\n" +
                "'username' = 'root',\n" +
                "'password' = '123456',\n"+
                "  'table-name' = 'test2',\n" +
                "  'connection.max-retry-timeout' = '60s'\n" +
                ")";
        tableEnv.executeSql(createOutputDDL);

        TableResult tableResult = tableEnv.executeSql("insert into ny_energy_data_target select * from ny_energy_data_source ");

        tableResult.print();

    }
}
